package com.polymathcoder.dpastry.core.util;

import com.polymathcoder.dpastry.api.Callback;
import com.polymathcoder.dpastry.api.CallbackRunnableFuture;
import com.polymathcoder.dpastry.api.environment.FlexProcessor;

import java.util.concurrent.*;

/**
 * TaskProcessor - Flexible Event(-driven model) Thread Pool Executor that implements the FlexProcessor interface.
 * <p/>
 * Extension of ThreadPoolExecutor that adds support for: (in no particular order)
 * 1) callbacks [implemented, draft]
 * 2) pause/resume for each runnable task via the Future [not yet implemented]
 * 3) scheduled executions? [not yet implemented, adapt ScheduledThreadPoolExecutor?]
 * 4) pause/resume for entire thread pool? [not yet implemented, but trivial (jdk example)]
 */
public class TaskProcessor extends ThreadPoolExecutor implements FlexProcessor {

    public TaskProcessor(int coreSize, int maxSize, long keepAliveTime) {
        //@TODO: Switch to PriorityBlockingQueue! (need to have an interface for Prioritized (& Comparable) Runnables!
        super(coreSize, maxSize, keepAliveTime, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
    }

    public TaskProcessor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    public TaskProcessor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
    }

    public TaskProcessor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
    }

    public TaskProcessor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
    }

    @Override
    @SuppressWarnings("unchecked")
    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);// not really needed (since super.afterExecture(r,t) is empty) but it's a good practice
        Object result = null;
        // apparently, in some cases, a future may not throw an error until you call the get() method... (cancel etc.)
        // so if we have no exception and a future, we should get the result and verify that there are no exceptions!
        if (t == null && r instanceof Future<?>) {
            try {
                result = ((Future<?>) r).get();
            } catch (CancellationException ce) {
                t = ce;
            } catch (ExecutionException ee) {
                t = ee.getCause();
            } catch (InterruptedException ie) {
                Thread.currentThread().interrupt(); // ignore/reset [propagate]
            }
        }
        // if we're working with a future that includes a callback, we should call the callback at this point with
        // the result of our execution
        if (r instanceof CallbackRunnableFuture && ((CallbackRunnableFuture) r).hasCallback()) {
            Callback c = ((CallbackRunnableFuture) r).getCallback();
            if (t != null) {
                c.onException(result, t);
            } else {
                c.onResult(result);
            }
        }
    }

    // ************ FLEX PROCESSOR API ************

    @Override
    public <V> Future<V> doTask(Callable<V> task) {
        return doTask(task, null);
    }

    @Override
    public <V> Future<V> doTask(Callable<V> task, Callback<V> callback) {
        return doTask(new CallbackFutureTask<V>(task, callback));
    }

    @Override
    public <V> Future<V> doTask(CallbackRunnableFuture<V> task) {
        execute(task);
        return task;
    }
}
